package org.offsync.async

import collection.JavaConversions._
import java.nio.channels.CancelledKeyException
import java.nio.channels.ClosedChannelException
import java.nio.channels.ClosedSelectorException
import java.nio.channels.{SelectionKey => JavaSelectionKey}
import java.nio.channels.{Selector => JavaSelector}
import java.nio.channels.spi.{SelectorProvider => JavaSelectorProvider}
import java.util.concurrent.atomic.AtomicInteger
import org.offsync.rich._
import org.offsync.util.CompletionException
import org.offsync.util.Disposable
import org.offsync.util.DisposableBase
import org.offsync.util.Disposer
import scala.collection._
import org.offsync.util.Trace.trace
import scala.collection.mutable.WeakHashMap
import org.offsync.util.Time

class AsyncSelector extends DisposableBase with Runnable {
  private val disposer = new Disposer()
  protected[async] val selector = java.nio.channels.spi.SelectorProvider.provider().openSelector()
  private val lock_ = new Object
  private var tasks = List[() => Unit]()
  private var running = true
  private val thread = disposer.dependsOn {
    val t = new Thread(this)
    t.start()
    t
  }
  
  def process(): Unit = {
    val count = selector.select()
    val taskList = lock_.synchronized { try tasks finally tasks = Nil }
    taskList.foreach { task =>
      try {
        task()
      } catch {
        case e: ClosedSelectorException => throw e.wrap()
        case e: CompletionException => throw e.wrap()
        case e => {
          println(e)
          e.printStackTrace(System.out)
          assert(false)
        }
      }
    }
    
    selector.selectedKeys.iterator.consume { key =>
      val selectable = key.attachment.asInstanceOf[AsyncSelectable]
      selectable.process(key) { () => }
      
      try {
        selectable.reregister(this);
      } catch {
        case e: ClosedChannelException => e.printStackTrace(System.err)
      }
    }
  }
  
  def run(): Unit = {
    try {
      while (isLive) {
        process()
      }
      
      closeAllInterests()
    } catch {
      case e: ClosedSelectorException => assert(false)
      case e: CompletionException => trace("completion exception")
    } finally {
      try selector.close() catch { case e => trace("close error") }
      
      super.onDispose()
    }
  }
  
  def closeAllInterests(): Unit = {
    for (task <- tasks) {
      task()
    }
    
    for (key <- selector.keys()) {
      val selectable = key.attachment().asInstanceOf[AsyncSelectable]
      selectable.closeInterestsFor(key, key.interestOps())
    }
  }
  
  def execute(f: => Unit): Unit = {
    val task = { () => f }
    lock_.synchronized {
      requireLive()
      tasks = task :: tasks
    }
    selector.wakeup()
  }
  
  def executeAlways(f: => Unit): Unit = {
    val task = { () => f }
    lock_.synchronized {
      tasks = task :: tasks
    }
    selector.wakeup()
  }
  
  protected override def onDispose(): Unit = {
    selector.wakeup()
  }
  
  override def waitDisposed(deadline: Time): Unit = {
    thread.join()
  } 
}
